package com.ccb.channel.sn;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ccb.filter.Filters;
import com.ccb.filter.RecordFilter;
import com.ccb.util.RecordType;
import com.ccb.util.StandardRecord;
import com.ccb.util.XmlJSON;
	
public class SnMapper extends Mapper<Object, Text, Text, Text> {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(SnMapper.class);
	private RecordFilter recordFilter ;
	private Text TRCD = new Text();
	private Text RECORD = new Text();
	private MultipleOutputs<Text, Text> mos = null;
	@Override
	protected void setup(Context context)
			throws IOException, InterruptedException {
		mos = new MultipleOutputs<Text, Text>(context);
		recordFilter = Filters.getFilter("sn");
		if(recordFilter == null ){
			logger.info("fiter is null");
		}
	}
	
	@Override
	protected void cleanup(Context context)
			throws IOException, InterruptedException {
		mos.close();
	}

	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("map(Object, Text, Context) - start");
		}

		String headerdata = "", xmldata = "";
		String values = value.toString().trim();
		if (values.isEmpty() || values.indexOf("{H:") > -1)
			return;
		try {
			StandardRecord record = new StandardRecord();
			values = values.substring(values.indexOf("{H:"));
			record.setRecordType(RecordType.REQUEST);
			int index = values.indexOf("<?xml");
			if (index > 0) {
				headerdata = values.substring(0, index);
				xmldata = values.substring(index);
			}

			if (headerdata.startsWith("{H:")) {
				String serviceId = StringUtils.substringAfterLast(headerdata, "XML").trim();
				if(recordFilter != null && !recordFilter.accept(serviceId)){
					return;
				}
				record.setServiceId(serviceId);
				TRCD.set(serviceId);
			}

			record.setRequest(xmldata);
			JSONObject obj = new JSONObject();
			obj = XmlJSON.xmlToJSONObj(xmldata);
			obj.put("headInfo", headerdata);
			record.setRequestData(obj);

			RECORD.set(JSON.toJSONString(record));
			mos.write(TRCD,RECORD, TRCD.toString());
			//context.write(TRCD, RECORD);
		} catch (Exception e) {
			logger.error("map(Object, Text, Context)", e);
			return;
		}
		if (logger.isDebugEnabled()) {
			logger.debug("map(Object, Text, Context) - end");
		}
		if (logger.isDebugEnabled()) {
			logger.debug("map(Object, Text, Context) - end");
		}
	}
}
